package com.csw.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.util.LongAccumulator
import org.apache.spark.{SparkConf, SparkContext}

object Demo18Acc {
  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName("acc")
      .setMaster("local")

    val sc: SparkContext = new SparkContext(conf)

    val studentRDD: RDD[String] = sc.textFile("spark/data/students.txt")

    /**
      * Executor：执行器，执行task，独立的JVM进程（Executor中的线程池去执行task）
      *
      * 算子中的代码会被封装成一个task发送到Executor执行
      * task为一个线程对象(包含代码逻辑)   task的数量由rdd的分区数量决定
      *
      * 算子外面的代码运行在Driver端(spark程序的主程序) Driver也是一个独立的JVM
      *
      * 因为不在一个JVM中所有i不变
      */

    /**
      * 算子内的代码运行在Executor，算子外面的代码运行在Driver端
      * 在算子内修改算子外的变量不会生效
      */
//    var i = 0
//    studentRDD.foreach(line => {
//      i += 1
//      println(line)
//    })
//
//    //i值为0
//    println(i)

    /**
      * 累加器：
      * 1、在每一个task中进行局部累加
      * 2、当job执行完成之后，在Driver端汇总
      */

    //定义一个累加器
    val accumulator: LongAccumulator = sc.longAccumulator

    studentRDD.foreach(line=>{
      //累加器可以在算子内进行累加
      accumulator.add(1)
      println(line)
    })

    //读取累加器的结果
    println(accumulator.value)


    //在spark的算子内不能再使用rdd
    //因为算子中的代码会被封装成一个task发送到Executor中执行，rdd本身就是一个弹性的分布式数据集，不能在网络中传输
//    studentRDD.foreach(i=>{
//      studentRDD.foreach(i=>{
//        println(1)
//      })
//    })
  }
}
